package com.gis507.mapreduce.done2;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.Utils.UtilsClass;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.junit.Test;

import java.io.IOException;

public class AISClassMapper extends Mapper<LongWritable, Text,Text, AisClassifyBean> {

    Text outK = new Text();
    AisClassifyBean outV = new AisClassifyBean();
    private int sum = 1;
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        if (sum == 1){

            sum += 1;

        }else {
            String line = value.toString();

            String[] comment = line.split(",");

            //String mmsi, String shipType, String nacStatusCN, double draught,
            // double heading, double course, double speed, String dest,
            // long unixTime, double lon_d, double lat_d, int seaRange
            String mmsi = comment[3];
            String shipType = comment[5];
            String nacStatusCN = comment[7];
            double length = UtilsClass.turnToDouble(comment[8]);
            double width = UtilsClass.turnToDouble(comment[9]);
            double draught = UtilsClass.turnToDouble(comment[10]);
            double heading = UtilsClass.turnToDouble(comment[11]);
            double course = UtilsClass.turnToDouble(comment[12]);
            double speed = UtilsClass.turnToDouble(comment[13]);
            String dest = comment[17];
            long unixTime = UtilsClass.turnToLong(comment[20]);
            double lon_d = UtilsClass.turnToDouble(comment[21]);
            double lat_d = UtilsClass.turnToDouble(comment[22]);
            //调用getTonnage方法算出吃水
            double tonnage = UtilsClass.getTonnage(shipType, length, width, draught);

            //fengzhuang
            outK.set(mmsi);

            outV.setMmsi(mmsi);
            outV.setShipType(shipType);
            outV.setNacStatusCN(nacStatusCN);
            outV.setLength(length);
            outV.setWidth(width);
            outV.setDraught(draught);
            outV.setHeading(heading);
            outV.setCourse(course);
            outV.setSpeed(speed);
            outV.setDest(dest);
            outV.setUnixTime(unixTime);
            outV.setLon_d(lon_d);
            outV.setLat_d(lat_d);
            outV.setTonnge(tonnage);

            //get the value of seaRange
            int seaRange = UtilsClass.getSeaRangeNumber(outV);

            //set seaRange for outV
            outV.setSeaRange(seaRange);

            //write for Mapper
            context.write(outK,outV);
        }
    }

    @Test
    public void test(){
//        String str1 = "";
//        String str2 = "23";
//
//        System.out.println(UtilsClass.turnToDouble(str1));
//        System.out.println(UtilsClass.turnToLong(str2));

        AisClassifyBean ais = new AisClassifyBean();
        ais.setLat_d(35.665205);
        ais.setLon_d(121.008173);

        AisClassifyBean newAis = UtilsClass.getNewAisClassifyBean(ais);

        System.out.println(ais + "---->" + newAis);
        System.out.println(ais.equals(newAis));

        int seaRangeNumber = UtilsClass.getSeaRangeNumber(ais);
        System.out.println(seaRangeNumber);
    }
}
